package com.processor.base;

import com.ProcessorThreadFactory;
import com.event.ProcessorEvent;
import com.handler.game.BaseHandler;
import com.handler.base.ThreadProcessorHandler;
import com.utils.GsonUtils;

import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author : gameZhao
 * @version V1.0
 * @Project: disruptorDemo
 * @Package com.processor.base
 * @Description: TODO
 * @date Date : 2021年12月27日 17:08
 */
public abstract class ThreadPoolProcessor extends BaseProcessor {
    
    
    protected final ThreadPoolExecutor executor;
    
    /**
     * 线程队列
     **/
    protected final BlockingQueue<Runnable> linkedBlockingQueue = new LinkedBlockingQueue<>();
    
    /**
     * 单线程
     *
     * @param name
     */
    public ThreadPoolProcessor(String name) {
        this(name, 1, 1, 0l, false);
    }
    
    /**
     * 固定大小线程池
     *
     * @param name        线程池名
     * @param threadCount 线程属相
     */
    public ThreadPoolProcessor(String name, int threadCount) {
        this(name, threadCount, threadCount, 0L, false);
    }
    
    /**
     * @param name
     * @param corePoolSize
     * @param maxPoolSize
     */
    public ThreadPoolProcessor(String name, int corePoolSize, int maxPoolSize) {
        this(name, corePoolSize, maxPoolSize, 0L, false);
    }
    
    /**
     * @param processorName
     * @param corePoolSize
     * @param maxPoolSize
     * @param keepTime
     * @param allowCoreThreadTimeOut
     */
    public ThreadPoolProcessor(String processorName, int corePoolSize, int maxPoolSize, long keepTime,
            boolean allowCoreThreadTimeOut) {
        // 线程名字
        this.processorName = processorName;
        
        // 线程实例
        this.executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepTime, TimeUnit.MILLISECONDS,
                linkedBlockingQueue, ProcessorThreadFactory.create(processorName));
        
        // 是否允许核心线程超时
        this.executor.allowCoreThreadTimeOut(allowCoreThreadTimeOut);
        
        // 启动所有核心线程，使其处于等待工作的空闲状态。仅当执行新任务时，此操作才重写默认的启动核心线程策略。
        this.executor.prestartAllCoreThreads();
    }
    
    @Override
    public int getQueueSize() {
        return this.linkedBlockingQueue.size();
    }
    
    @Override
    public String getQueueInfo() {
        String info = "[]";
        HashMap<String, Integer> map = new HashMap<String, Integer>();
        for (Runnable runnable : this.linkedBlockingQueue) {
            ThreadProcessorHandler threadPoolProcessorHandler = (ThreadProcessorHandler) runnable;
            String name = threadPoolProcessorHandler.getProcessorEvent().getClass().getSimpleName();
            if (map.containsKey(name)) {
                map.put(name, map.get(name) + 1);
            } else {
                map.put(name, 1);
            }
        }
        if (map.size() > 0) {
            info = GsonUtils.toJson(map);
        }
        return info;
    }
    
    /**
     * 向该线程投递Handler(Runnable)
     *
     * @param handler
     */
    @Override
    public void executeHandler0(BaseHandler handler) {
        try {
            if (executor.isShutdown()) {
                return;
            }
            handler.setTime(System.currentTimeMillis());
            ThreadProcessorHandler threadPoolProcessorHandler = new ThreadProcessorHandler(
                    new ProcessorEvent(handler));
            executor.execute(threadPoolProcessorHandler);
        } catch (Exception e) {
        }
    }
    
    
    /**
     * 线程是否已关闭
     * <p>
     * 如果isCompleteAllCommand为true, 则已经执行过stop, 并且已提交的命令全部执行完的情况下返回true
     *
     * @param isCompleteAllCommand 是否完成所有任务
     * @return
     */
    @Override
    public boolean isStop(boolean isCompleteAllCommand) {
        if (isCompleteAllCommand) {
            return executor.isTerminated();
        } else {
            return executor.isShutdown();
        }
    }
    
    /**
     * 关闭线程池
     *
     * @param await 是否等待队列全部执行完成
     */
    @Override
    public void stop(boolean await) {
        if (await) {
            executor.shutdown();
        } else {
            executor.shutdownNow();
        }
        
        try {
            while (!this.executor.isTerminated()) {
                this.executor.awaitTermination(1L, TimeUnit.SECONDS);
            }
        } catch (Exception e) {
        }
    }
}
